package com.atguigu.flink.state.keyedstate;

import com.atguigu.flink.func.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by 黄凯 on 2023/6/20 0020 21:10
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 计算每种传感器的水位和：
 * *          输入： Integer
 * *          输出:  Integer
 * *
 * *          输入和输出是一样的类型用ReducingState
 * *
 * *  计算每种传感器的平均水位
 * *          输入： Integer
 * *          输出:  Double
 * *
 * *          输入和输出不一样的类型用AggegatingState
 */
public class Flink10_AggeragatingState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
                .socketTextStream("127.0.0.1", 8888)
                .map(new WaterSensorMapFunction());

        KeyedStream<WaterSensor, String> keyedDS = ds.keyBy(WaterSensor::getId);

        keyedDS
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    private AggregatingState<Integer, Double> avgVc;

                    //初始化方法
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //找到状态的备份存储,取出状态并赋值
                        avgVc = getRuntimeContext().getAggregatingState(new AggregatingStateDescriptor<>("avgVc",
                                new AggregateFunction<Integer, MyAcc, Double>() {
                                    @Override
                                    public MyAcc createAccumulator() {
                                        return new MyAcc(0, 0d);
                                    }

                                    @Override
                                    public MyAcc add(Integer value, MyAcc accumulator) {
                                        accumulator.count += 1;
                                        accumulator.sumVc += value;
                                        return accumulator;
                                    }

                                    @Override
                                    public Double getResult(MyAcc accumulator) {
                                        return accumulator.sumVc / accumulator.getCount();
                                    }

                                    //BatchAPI,Streaming无需写
                                    @Override
                                    public MyAcc merge(MyAcc a, MyAcc b) {
                                        return null;
                                    }
                                }, MyAcc.class));
                    }


                    @Override
                    public void processElement(WaterSensor waterSensor, KeyedProcessFunction<String, WaterSensor, String>.Context context, Collector<String> collector) throws Exception {

                        //只要存进去，自动聚合
                        avgVc.add(waterSensor.getVc());

                        collector.collect(context.getCurrentKey() + " avgVc:" + avgVc.get());

                    }
                }).print();

        env.execute();

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc {
        private Integer count;
        private Double sumVc;
    }

}
